package kafka

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/Shopify/sarama"
)

// NewDelayQueue 创建一个延迟队列
func NewDelayQueue(addrs []string, delayTime time.Duration, topic string) *DelayQueueProducer {
	delayTopic := topic + "_delay_queue"

	consumerConfig := sarama.NewConfig()
	consumerConfig.Version = sarama.V2_0_1_0
	//consumerConfig.Consumer.Offsets.AutoCommit.Enable = false
	//consumerConfig.Consumer.Group.Rebalance.Timeout = 60 * time.Second
	delayServiceConsumerGroup, err := sarama.NewConsumerGroup(addrs, topic+"_delay_group", consumerConfig)
	if err != nil {
		log.Fatal(err)
	}

	delayServiceProducerClient, err := NewSyncProducerClient(addrs, func(config *sarama.Config) {
		config.Producer.Flush.Frequency = 500 * time.Millisecond  // 每0.5s冲洗一批
		config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follow都确认
		config.Producer.Partitioner = sarama.NewRandomPartitioner // 设置生产者的数据写入哪个分区（此处随机设置一个分区）
	})
	if err != nil {
		log.Fatal(err)
	}

	// 启动延迟服务
	delayServiceConsumer := NewDelayServiceConsumer(delayServiceProducerClient, delayTime, topic)
	go func() {
		for {
			if err := delayServiceConsumerGroup.Consume(context.Background(),
				[]string{delayTopic}, delayServiceConsumer); err != nil {
				break
			}
		}
	}()

	//if err := delayServiceConsumerGroup.Consume(context.Background(), []string{delayTopic}, delayServiceConsumer); err != nil {
	//	log.Fatal(err)
	//}

	return &DelayQueueProducer{
		producer:   delayServiceProducerClient,
		delayTopic: delayTopic,
	}
}

type DelayQueueProducer struct {
	producer   sarama.SyncProducer // 生产者
	delayTopic string              // 延迟服务主题
}

// SendMessage 发送消息
func (q *DelayQueueProducer) SendMessage(value string) (partition int32, offset int64, err error) {
	return q.producer.SendMessage(&sarama.ProducerMessage{
		Topic:     q.delayTopic,
		Value:     sarama.ByteEncoder(value),
		Timestamp: time.Now(),
	})
}

func NewDelayServiceConsumer(producer sarama.SyncProducer, delay time.Duration,
	realTopic string) *DelayServiceConsumer {
	return &DelayServiceConsumer{
		producer:  producer,
		delay:     delay,
		realTopic: realTopic,
	}
}

// DelayServiceConsumer 延迟服务消费者
type DelayServiceConsumer struct {
	producer  sarama.SyncProducer
	delay     time.Duration
	realTopic string
}

func (c *DelayServiceConsumer) ConsumeClaim(session sarama.ConsumerGroupSession,
	claim sarama.ConsumerGroupClaim) error {
	for message := range claim.Messages() {
		fmt.Println(string(message.Value))
		// 如果消息已经超时，把消息发送到真实队列
		now := time.Now()
		fmt.Println(now.Sub(message.Timestamp))
		if now.Sub(message.Timestamp) >= c.delay {
			_, _, err := c.producer.SendMessage(&sarama.ProducerMessage{
				Topic: c.realTopic,
				Key:   sarama.ByteEncoder(message.Key),
				Value: sarama.ByteEncoder(message.Value),
			})
			fmt.Println(111)
			if err == nil {
				session.MarkMessage(message, "")
				continue
			}
		}
		session.MarkOffset(message.Topic, message.Partition, message.Offset, "")
		time.Sleep(time.Second)
		return nil
	}
	return nil
}

func (c *DelayServiceConsumer) Setup(sarama.ConsumerGroupSession) error {
	return nil
}

func (c *DelayServiceConsumer) Cleanup(sarama.ConsumerGroupSession) error {
	return nil
}
